package com.bigdata.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 间隔5s执行一次
 * 微批，指定间隔时间执行一次RDD计算
 */
object Demo1WordCount {
  def main(args: Array[String]): Unit = {

    /**
     * 创建spark Streaming环境,需要依赖Spark core环境
     */
    val conf = new SparkConf()

    //指定2个core执行代码
    conf.setMaster("local[2]")

    conf.setAppName("wc")

    val sc = new SparkContext(conf)

    sc.setLogLevel("error")

    /**
     * 创建spark streaming环境，需要设置执行的间隔时间
     * 5秒计算一次，计算的数据就是最近5秒产生的数据
     */
    val ssc = new StreamingContext(sc, Durations.seconds(5))

    /**
     * 1、读取实时的数据源
     * 生产环境接入的数据源是kafka
     * 安装
     * yum install nc
     * 启动socket服务
     * nc -lk 8888
     */
    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("node01", 8888)

    /**
     * 统计单词的数量
     * hello,word
     * hello,hadoop
     * hello,spark
     */
//    val wordsDS: DStream[String] = linesDS.flatMap(_.split(","))
//
//    val kvDS: DStream[(String, Int)] = wordsDS.map((_, 1))
//
//    val countDS: DStream[(String, Int)] = kvDS.reduceByKey(_ + _)

    /**
     * 打印结果
     *
     */
    linesDS.print()


    /**
     * 启动spark streaming程序
     */
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }

}
